
package com.example.paper.common.datam.migrate.processor;

import com.example.paper.common.datam.constant.KeyConstant;
import com.example.paper.common.datam.migrate.job.JobCache;
import com.example.paper.common.datam.migrate.job.TraceIdHolder;
import com.example.paper.common.datam.migrate.reader.CommonBlock;
import com.example.paper.common.datam.migrate.reader.Reader;
import com.example.paper.common.datam.migrate.writer.Writer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorCompletionService;

/**
 * 描述.
 *
 * @author Long Fengfan, 2022年10月13日

 */

@Service
@Slf4j
public abstract class BaseProcessor<R, W> implements Processor<R, W> {

    @Override
    public void migrate(Reader<R> reader, Writer<W> writer, Map<String, Object> processorParams) {
        Map<String, Object> readerParams = new HashMap<>(processorParams);
        long totalCount = reader.getTotalCount(readerParams);
        int batchSize = reader.getBatchSize();
        List<CommonBlock> commonBlocks = getCommonBlocks(totalCount, batchSize);
        ExecutorCompletionService<CommonBlock> completionService = new ExecutorCompletionService<>(DataThreadPool.executor);
        String jobId = TraceIdHolder.getJobId();
        Integer nextProcessId = TraceIdHolder.getNextProcessId();
        processorParams.put(KeyConstant.job_id, jobId);
        processorParams.put(KeyConstant.process_id, nextProcessId);
        log.info("jobId:{},processId:{}", jobId, nextProcessId);
        Map<Integer, Boolean> processErrorMap = new HashMap<>();
        JobCache.isErrorCache.putIfAbsent(jobId, processErrorMap);
        processErrorMap.putIfAbsent(nextProcessId, false);

        for (CommonBlock commonBlock : commonBlocks) {
            completionService.submit(() -> {
                List<R> read = reader.read(readerParams, processorParams, commonBlock);
                List<W> wList = transfer(read);
                writer.write(processorParams, wList);
                log.info("线程:{}, begin:{},end:{} 完成", Thread.currentThread().getName(), commonBlock.getBegin(), commonBlock.getEnd());
                return commonBlock;
            });
        }
        for (int size = commonBlocks.size(); size > 0; size--) {
            try {
                CommonBlock block = completionService.take().get();
            } catch (Exception e) {
                log.info("参数:{}, 失败", processorParams);
                processErrorMap.put(nextProcessId, true);
                e.printStackTrace();
                rollBack(processorParams);
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 批量转换
     *
     * @param rList rList
     * @return List<W>
     */
    public List<W> transfer(List<R> rList) {
        List<W> ws = new ArrayList<>();
        for (R r : rList) {
            W w = transfer(r);
            ws.add(w);
        }
        return ws;
    }

    /**
     * 需要实现具体单个转换
     *
     * @param r
     * @return
     */
    public abstract W transfer(R r);


    private List<CommonBlock> getCommonBlocks(long total, int blockSize) {
        long totalPages = (BigDecimal.valueOf(Math.ceil((double) total * 1.0D / (double) blockSize))).longValue();
        ArrayList<CommonBlock> commonBlocks = new ArrayList<>();
        for (int i = 0; (long) i < totalPages; ++i) {
            long begin = i * (long) blockSize;
            long end = begin + (long) blockSize;
            end = Math.min(end, total);
            CommonBlock b = new CommonBlock(total, i * blockSize, Math.min(end, total), i + 1, (int) (end - begin), blockSize);
            commonBlocks.add(b);
        }
        return commonBlocks;
    }
}
